package com.sh.data.engine.application.service.integration;

import cn.hutool.core.convert.Convert;
import com.google.common.collect.Lists;
import com.sh.data.engine.application.model.dto.base.PagerResponse;
import com.sh.data.engine.application.model.dto.common.DirectoryResponse;
import com.sh.data.engine.application.model.dto.integration.*;
import com.sh.data.engine.application.model.mapper.integration.FlinkDataSyncTaskPageResponsePageMapper;
import com.sh.data.engine.application.model.mapper.integration.FlinkDatabaseResponseMapper;
import com.sh.data.engine.application.model.request.common.UpsertDirectoryRequest;
import com.sh.data.engine.application.model.request.integration.DataSourceSaveRequest;
import com.sh.data.engine.application.model.request.integration.FlinkDataSyncTaskRequest;
import com.sh.data.engine.application.model.request.integration.PreviewDataForMqRequest;
import com.sh.data.engine.application.model.request.integration.PreviewJsonDataForSpRequest;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.enums.DirectoryMenuEnum;
import com.sh.data.engine.domain.common.model.domain.DirectoryDomain;
import com.sh.data.engine.domain.common.model.param.UpsertDirectoryParam;
import com.sh.data.engine.domain.common.service.DirectoryService;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.datadev.offline.model.domain.OnlineTaskLogDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceCreateTableDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDomain;
import com.sh.data.engine.domain.integration.datasource.model.param.DataSourceSaveParam;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.online.model.domain.*;
import com.sh.data.engine.domain.integration.online.model.param.PreviewDataForMqParam;
import com.sh.data.engine.domain.integration.online.model.param.StartOnlineKafkaTaskParam;
import com.sh.data.engine.domain.integration.online.service.FlinkDataSyncService;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Set;

/**
 * @author: mengzheng.mhc
 * @date: 2024/7/2 15:26
 */
@Slf4j
@Service
public class FlinkDataSyncApplicationService {

    @Autowired
    private FlinkDataSyncService flinkDataSyncService;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private DirectoryService directoryService;

    @Autowired
    private TblPropService tblPropService;

    @Autowired
    private FlinkDatabaseResponseMapper flinkDatabaseResponseMapper;

    @Autowired
    private FlinkDataSyncTaskPageResponsePageMapper flinkDataSyncTaskPageResponsePageMapper;

    public boolean insert(FlinkDataSyncTaskRequest request) {
        FlinkDataSyncTaskDomain domain = Convert.convert(FlinkDataSyncTaskDomain.class, request);
        flinkDataSyncService.insert(domain);
        return true;
    }

    public boolean publish(Long taskId) {
        StartOnlineKafkaTaskParam param = new StartOnlineKafkaTaskParam();
        param.setTaskId(taskId);
        flinkDataSyncService.publish(param);
        return true;
    }

    public OnlineTaskLogResponse getTaskLog(Long jobId, Integer offset) {
        OnlineTaskLogDomain taskLog = flinkDataSyncService.getTaskLog(jobId, offset);

        return ConvertUtil.copyProperties(taskLog, OnlineTaskLogResponse.class);
    }

    public boolean reRun(Long taskId) {
        flinkDataSyncService.reRun(taskId);
        return true;
    }

    public boolean unPublish(Long taskId) {
        flinkDataSyncService.unPublish(taskId);
        return true;
    }

    public boolean update(FlinkDataSyncTaskRequest request) {
        FlinkDataSyncTaskDomain domain = Convert.convert(FlinkDataSyncTaskDomain.class, request);
        flinkDataSyncService.update(domain);
        return true;
    }

    public boolean delete(Long taskId) {
        flinkDataSyncService.delete(taskId);
        return true;
    }

    public List<FlinkDatabaseResponse> getDataBaseByDirection(Integer direction, String dsType) {
        List<FlinkDatabaseDomain> list = flinkDataSyncService.getDataBaseByDirection(direction, dsType);
        return flinkDatabaseResponseMapper.map(list);
    }

    public List<FlinkDatabaseResponse> getListByDsTypes(Long projectId) {
        var dsTypes = Lists.newArrayList("MySQL", "TiDB", "OpenTSDB", "IoTDB");
        List<DataSourceDomain> list = dataSourceService.getListByDsTypes(projectId, dsTypes);
        return ConvertUtil.copyProperties(list, FlinkDatabaseResponse.class);
    }

    public List<FlinkTableResponse> getTables(Long dsId, String schema, Integer direction) {
        List<FlinkTableDomain> tables = flinkDataSyncService.getTables(dsId, schema, direction);
        return Convert.toList(FlinkTableResponse.class, tables);
    }

    public List<FlinkTableFieldResponse> getTableFields(Long dsId, String tableName) {
        List<FlinkTableFieldDomain> tableFields = flinkDataSyncService.getTableFields(dsId, tableName);
        return Convert.toList(FlinkTableFieldResponse.class, tableFields);
    }

    public Set<String> getTopics(Long dsId) {
        return flinkDataSyncService.getTopics(dsId);
    }

    public PagerResponse<FlinkDataSyncTaskPageResponse> getFlinkTaskPage(Integer pageNum, Integer pageSize, String taskName, Integer taskStatus, Long dirId, Integer taskRunStatus, Date startTime, Date endTime) {
        FlinkDataSyncTaskQueryDomain offlineSyncPageQueryDomain =
            FlinkDataSyncTaskQueryDomain.builder()
                .taskName(taskName)
                .taskStatus(taskStatus)
                .pageNum(pageNum)
                .pageSize(pageSize)
                .dirId(dirId)
                .taskRunStatus(taskRunStatus)
                .publishDate(startTime)
                .endTime(endTime)
                .build();
        PageResult<FlinkDataSyncTaskPageDomain> pageDomain =
            flinkDataSyncService.getPageList(offlineSyncPageQueryDomain);
        return flinkDataSyncTaskPageResponsePageMapper.toPageResponse(pageDomain);
    }

    public FlinkTaskSummaryResponse getSummary() {
        FlinkTaskSummaryDomain summary = flinkDataSyncService.getSummary();
        return ConvertUtil.copyProperties(summary, FlinkTaskSummaryResponse.class);
    }

    public OnlinePreviewDataResponse previewData(Long dsId, String tableName) {
        PreviewDataDomain previewDataDomain = dataSourceService.previewData(dsId, tableName);
        return Convert.convert(OnlinePreviewDataResponse.class, previewDataDomain);
    }

    public String createTargetTable(Long dsId, String tableName, Long targetDsId, String schema, int infoCase) {
        DataSourceDetailDomain dataSource = dataSourceService.getDataSourceDetailById(dsId, false);
        if (DSType.Kafka.name().equalsIgnoreCase(dataSource.getDsType())
            || DSType.MQTT.name().equalsIgnoreCase(dataSource.getDsType())) {
            throw new BusinessException("来源表类型为Kafka、MQTT，不支持一键建表");
        }

        return dataSourceService.getCreateTableSql(dsId, tableName, targetDsId, schema, infoCase);
    }

    public List<List> previewDataForMq(PreviewDataForMqRequest request) {
        PreviewDataForMqParam param = Convert.convert(PreviewDataForMqParam.class, request);
        return flinkDataSyncService.previewDataForMq(param);
    }

    public List<List> previewJsonDataForSp(PreviewJsonDataForSpRequest request) {
        PreviewDataForMqParam param = Convert.convert(PreviewDataForMqParam.class, request);
        return flinkDataSyncService.previewDataForMq(param);
    }

    public OfflineSyncCreateTableResponse executeCreateTableSql(Long dsId, String scripts, Long dataSecretId) {
        // 设置了密级
        boolean requireSetDataSecret = Objects.nonNull(dataSecretId);
        long secretId = -1;
//        if (requireSetDataSecret) {
//            secretId = Long.valueOf(dataSecretId);
//            Long securityLevelId = UserContext.getUserInfo().getSecurityLevelId();
//            log.info("user secretId: {}, data secretId: {}", securityLevelId, secretId);
//            // 判断当前用户密级
//            if (!secretMgtService.validSecretLevel(securityLevelId, secretId)) {
//                return RestResponseEntity.error("数据密级设置异常");
//            }
//        }
        DataSourceCreateTableDomain dataSourceCreateTableDomain =
            dataSourceService.executeCreateTableSql(dsId, scripts);

        OfflineSyncCreateTableResponse offlineSyncCreateTableResponse =
            ConvertUtil.copyProperties(
                dataSourceCreateTableDomain, OfflineSyncCreateTableResponse.class);
        Long projectId = ServletUtils.getProjectId();
//        if (requireSetDataSecret) {
        TblPropEntity tblPropEntity = new TblPropEntity();
        tblPropEntity.setProjectId(projectId);
        tblPropEntity.setSchema(dataSourceCreateTableDomain.getSchema());
        tblPropEntity.setTblId(tblPropService.generateTblId(projectId));
        tblPropEntity.setDataSecretId(secretId);
        tblPropEntity.setTableName(dataSourceCreateTableDomain.getOnlyTableName());
        tblPropEntity.setLifecycle(-1);
        tblPropEntity.setTblType(TableInfoDomain.TblType.MANAGED_TABLE.name());
        tblPropEntity.setRowState(1);
        tblPropService.save(tblPropEntity);
//        }

        return offlineSyncCreateTableResponse;
    }

    public FlinkTaskDetailInfoResponse getDetailInfo(Long taskId) {
        FlinkDataSyncTaskDomain domain = flinkDataSyncService.getDetailInfo(taskId);
        return ConvertUtil.copyProperties(domain, FlinkTaskDetailInfoResponse.class);
    }

    public List<DirectoryResponse> getDirList(String search) {
        Long projectId = ServletUtils.getProjectId();
        List<DirectoryDomain> dirList = directoryService.getDirList(projectId, DirectoryMenuEnum.ONLINE, search);
        return ConvertUtil.copyProperties(dirList, DirectoryResponse.class);
    }

    public boolean upsetDir(UpsertDirectoryRequest request) {
        Long projectId = ServletUtils.getProjectId();
        UpsertDirectoryParam param = Convert.convert(UpsertDirectoryParam.class, request);
        param.setProjectId(projectId);
        param.setMenuCode(DirectoryMenuEnum.ONLINE.getCode());

        return directoryService.upsetDir(param);
    }

    public boolean deleteDir(Long id) {
        return flinkDataSyncService.deleteDir(id);
    }

    public List<String> getSchemas(Long dsId, String dsType) {
        return dataSourceService.getDataSourceSchema(dsId, dsType);
    }

    public Long checkSpKafka(DataSourceSaveRequest dataSourceSaveRequest) {
        Long projectId = ServletUtils.getProjectId();
        Long dsId = dataSourceService.checkSpKafka(projectId);
        if (dsId == null) {
            DataSourceSaveParam dataSourceParam =
                ConvertUtil.copyProperties(dataSourceSaveRequest, DataSourceSaveParam.class);
            dataSourceParam.setProjectId(projectId);
            dsId = dataSourceService.saveDataSource(dataSourceParam);
        }
        return dsId;
    }
}
